package com.caption.mq.core;

import com.alibaba.fastjson.JSONObject;
import com.caption.mq.entity.MessageDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;
import java.nio.charset.StandardCharsets;

/**
 * @description:
 * @author: 码农
 * @date: 2022-01-16 15:33:19
 */
@Slf4j
public class MqUtils {
    private DefaultMQProducer defaultMQProducer;

    private MessageQueueSelector messageQueueSelector = new SelectMessageQueueByHash();

    public DefaultMQProducer getDefaultMQProducer() {
        return defaultMQProducer;
    }

    public void setDefaultMQProducer(DefaultMQProducer defaultMQProducer) {
        this.defaultMQProducer = defaultMQProducer;
    }

    public void setMessageQueueSelector(MessageQueueSelector messageQueueSelector) {
        this.messageQueueSelector = messageQueueSelector;
    }

    /**
     * 发送普通消息
     * @param messageDTO
     */
    public void send(MessageDTO messageDTO) throws Exception {
        send(messageDTO,null);
    }

    public void send( MessageDTO messageDTO, SendCallback sendCallback) throws Exception {
        String topic = messageDTO.getTopic();
        String  tag = messageDTO.getTag();
        if(StringUtils.isEmpty(topic)){
          throw new RuntimeException("消息投递topic不能为空");
        }
        //不传默认全部
        if(StringUtils.isEmpty(tag)){
            tag ="*";
        }
        SendCallback callback = sendCallback == null ? new DefaultSendCallback() : sendCallback;
        Message message = new Message();
        message.setTopic(topic);
        message.setTags(tag);
        String data = JSONObject.toJSONString(messageDTO);
        message.setBody(data.getBytes(StandardCharsets.UTF_8));
        this.defaultMQProducer.send(message, callback);
    }

    /**
     * 发送顺序消息
     */
    public void sendOrderly(String hashKey, MessageDTO messageDTO) throws Exception {
        sendOrderly(hashKey, messageDTO, null);
    }

    public void sendOrderly(String hashKey, MessageDTO messageDTO, SendCallback sendCallback) throws Exception {
        String topic = messageDTO.getTopic();
        String  tag = messageDTO.getTag();
        if(StringUtils.isEmpty(topic)){
            throw new RuntimeException("消息投递topic不能为空");
        }
        //不传默认全部
        if(StringUtils.isEmpty(tag)){
            tag ="*";
        }
        SendCallback callback = sendCallback == null ? new DefaultSendCallback() : sendCallback;
        Message message = new Message();
        message.setTopic(topic);
        message.setTags(tag);
        String data = JSONObject.toJSONString(messageDTO);
        message.setBody(data.getBytes(StandardCharsets.UTF_8));
        this.defaultMQProducer.send(message, messageQueueSelector, hashKey, callback);
    }

    private static class DefaultSendCallback implements SendCallback {
        @Override
        public void onSuccess(SendResult sendResult) {
            log.info("发送消息成功{}", sendResult);
        }

        @Override
        public void onException(Throwable e) {
            log.info("发送消息失败{}", e);
        }
    }


}
